package com.duowan.realtime.computing;

import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import org.apache.thrift.TException;
import org.springframework.util.Assert;

import com.duowan.realtime.computing.util.BeanFactory;
import com.duowan.realtime.thirft.api.HyperLogLogException;
import com.duowan.realtime.thirft.api.HyperLogLogPlusException;
import com.duowan.realtime.thirft.api.HyperLogLogPlusQuery;
import com.duowan.realtime.thirft.api.HyperLogLogQuery;
import com.duowan.realtime.webservice.DataComputingWebService;

public class HyperLogLogClient {
	
	private final static Map<String,RawHyperLogLogClient> hyperloglogGroupMap = new ConcurrentHashMap<String, RawHyperLogLogClient>();
	
	private static DataComputingWebService dataComputingWebService;
	
	public HyperLogLogClient(){
		dataComputingWebService = BeanFactory.getDataComputingWebService();
	}
	
	/**
	 * 此方法用于计算历史新用户
	 * @param hllGroup hyperloglog组
	 * @param hllpQueryList	批量的hyperloglogplus model
	 * @return values在该group下含有多少个新增
	 */
//	public Map<String, Integer> offerForCardinality(String hllGroup, List<HyperLogLogQuery> hllQueryList) throws HyperLogLogPlusException, TException {
//		return doComputing(hllGroup, hllQueryList);
//	}
	
	public void offer(String hllGroup, List<HyperLogLogQuery> hllQueryList) throws HyperLogLogPlusException, TException {
		RawHyperLogLogClient client = initClient(hllGroup);
		client.offer(hllGroup, hllQueryList);
	}

	public long cardinality(String hllGroup, String hllQueryGroup) throws HyperLogLogException, TException{
		RawHyperLogLogClient client = initClient(hllGroup);
		return client.cardinality(hllGroup, hllQueryGroup);
	}

	public List<Map<String, String>> keys(String hllGroup, String key) throws HyperLogLogException, TException{
		RawHyperLogLogClient client = initClient(hllGroup);
		return client.keys(hllGroup, key);
	}
	
	private RawHyperLogLogClient initClient(String hllGroup) {
		return getClientByHllGroup(hllGroup);
	}
	
	private RawHyperLogLogClient getClientByHllGroup(String hllpGroup) {
		RawHyperLogLogClient client = hyperloglogGroupMap.get(hllpGroup);
		if(client == null) {
			String serverIp = dataComputingWebService.getServerIp(hllpGroup);
			Assert.hasText(serverIp,"not found serverIp for hllpGroup:"+hllpGroup);
			client = new RawHyperLogLogClient();
			client.setHost(serverIp);
			try {
				client.afterPropertiesSet();
			} catch (Exception e) {
				new RuntimeException("afterPropertiesSet error:", e);
			}
			hyperloglogGroupMap.put(hllpGroup, client);
		}
		return client;
	}


}
